package com.wangwenjun.concurrency.chapter14.threadpool;

import java.sql.Time;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @Package: com.wangwenjun.concurrency.chapter14.threadpool
 * @ClassName: UseThreadPool
 * @Author: lujieni
 * @Description: 1
 * @Date: 2021-03-11 14:26
 * @Version: 1.0
 */
public class UseThreadPool {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        testInvokeAll();
    }


    /**
     * threadPoolExecutor.invokeAll(tasks)
     *
     * each of which has completed 返回future的时候就代表任务已经执行完成,而不是在e.get()时阻塞
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void testInvokeAll() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                3,
                3,
                60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());
        List<Callable<Integer>> tasks = IntStream.of(1, 10, 20).boxed().map(i -> (Callable<Integer>) () -> {
            System.out.println(Thread.currentThread().getName() + ":" + i);
            TimeUnit.SECONDS.sleep(i);
            return i;
        }).collect(Collectors.toList());

        threadPoolExecutor.invokeAll(tasks).stream().forEach(e->{
            try {
                Integer i = e.get();
                System.out.println(Thread.currentThread().getName()+" get result:"+i);
            } catch (InterruptedException ex) {
                throw new RuntimeException(ex);
            } catch (ExecutionException ex) {
                throw new RuntimeException(ex);
            }
        });

    }


    /**
     *
     * tasks that have not completed are cancelled
     * 其余没有执行完成的任务将会被打断(前提是代码里加了打断的校验否则没办法中断的)
     *
     * Upon normal or exceptional return, tasks that have not completed are cancelled.
     * @throws ExecutionException
     * @throws InterruptedException
     */

    public static void testInvokeAny() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                3,
                3,
                60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());
        List<Callable<Integer>> tasks = IntStream.range(1, 4).boxed().map(i -> (Callable<Integer>) () -> {
            if(i != 1){
                while (!Thread.interrupted()){
                    System.out.println("working");
                }
            }else{
                TimeUnit.SECONDS.sleep(i);
            }
            System.out.println(Thread.currentThread().getName() + ":" + i);
            return i;
        }).collect(Collectors.toList());

        Integer i = threadPoolExecutor.invokeAny(tasks);
        System.out.println(i);

    }




    /**
     * 线程池满了就把阻塞队列的队头干掉,自己放到队尾去
     */
    public static void testDiscardOldestPolicy() {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                1,
                2,
                60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardOldestPolicy());

        for (int i = 0; i < 3; i++) {
            threadPoolExecutor.submit(()->{
                try {
                    TimeUnit.SECONDS.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

            });
        }

        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        threadPoolExecutor.execute(()->{
            System.out.println(Thread.currentThread().getName());
        });

        System.out.println("====================");
    }



    /**
     * 线程池满了那就当前提交任务的线程执行吧
     */
    public static void testCallerRunsPolicy() {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                1,
                2,
                60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());

        for (int i = 0; i < 3; i++) {
            threadPoolExecutor.submit(()->{
                try {
                    TimeUnit.SECONDS.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

            });
        }

        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        threadPoolExecutor.execute(()->{
            System.out.println(Thread.currentThread().getName());
        });

        System.out.println("====================");
    }



    /**
     * DiscardPolicy 满了就直接抛弃这个任务
     */
    public static void testDiscardPolicy() {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                1,
                2,
                60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardPolicy());

        for (int i = 0; i < 3; i++) {
            threadPoolExecutor.submit(()->{
                try {
                    TimeUnit.SECONDS.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

            });
        }

        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        threadPoolExecutor.execute(()->{
            System.out.println(Thread.currentThread());
        });

        System.out.println("====================");
    }


    /**
     * AbortPolicy submit的那个线程直接抛异常
     * @throws InterruptedException
     */
    public static void testAbortPolicy() throws InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                1,
                2,
                60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());

        for (int i = 0; i < 3; i++) {
            threadPoolExecutor.submit(()->{
                try {
                    TimeUnit.SECONDS.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

            });
        }

        TimeUnit.SECONDS.sleep(2);

        threadPoolExecutor.execute(()->{
            System.out.println(Thread.currentThread());
        });

        System.out.println("====================");
    }


    /**
     *  shutdown() 之后新的任务拒绝执行,老任务还会继续完成执行再结束
     *
     */
    public static void test01(){
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.execute(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        System.out.println(executorService.isShutdown()); // false
        executorService.shutdown();

        executorService.execute(()->{
            System.out.println("new Task"); // main线程会抛异常,因为线程池已经shutdown了
        });

        System.out.println(executorService.isShutdown()); // true
        System.out.println(executorService.isTerminated()); // false
    }

    /**
     * 使用线程池的时候自己包装task自己捕获异常就好啦
     */
    public static void test02(){
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.execute(new MyTask() {
            @Override
            void doBusiness() {
                int j = 5 / 0;
            }

            @Override
            void solveExcepetion(Throwable e) {
                System.out.println("solveExcepetion");
                e.printStackTrace();
            }
        });



    }

    abstract private static class MyTask implements Runnable{

        @Override
        public void run() {
            try{
                doBusiness();
            }catch (Throwable e){
                solveExcepetion(e);
            }
        }

        abstract void doBusiness();

        abstract void solveExcepetion(Throwable e);

    }




    public static void test12(){
        //ExecutorService executorService = Executors.newSingleThreadExecutor();//单个线程
        //ExecutorService executorService = Executors.newFixedThreadPool(5);//创建一个固定的线程池的大小
        ExecutorService executorService = Executors.newCachedThreadPool();//可伸缩的,遇强则强

        for(int i = 0;i < 100;i++){
            executorService.execute(()->{
                System.out.println(Thread.currentThread().getName()+" ok");
            });
        }
        executorService.shutdown();
    }


}